package com.xzj.common.plugins.disruptor.util;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.xzj.common.plugins.disruptor.domain.DisruptorMessage;
import com.xzj.common.plugins.disruptor.factory.DisruptorEventFactory;
import com.xzj.common.plugins.disruptor.factory.NameThreadFactory;
import com.xzj.common.plugins.disruptor.mq.DisruptorConsumer;
import com.xzj.common.plugins.disruptor.mq.DisruptorMqProducer;
import lombok.extern.slf4j.Slf4j;

import java.util.function.Consumer;

/**
 * @program: xzj-common
 * @ClassName: DisruptorUtil
 * @description: Disruptor内部消息队列bean创建方法
 * @author: XiaZunJun
 * @create: 2024-11-19 10:45
 */
@Slf4j
public class DisruptorUtil<T> {

    private final DisruptorMqProducer<T> disruptorMqProducer;

    /**
     * 创建消息队列方法
     *
     * @param threadName 消费线程名称前缀
     * @param handleFun 消息消费函数
     * @author xiazunjun
     * @date 2023/8/29 14:50
     */
    public DisruptorUtil(String threadName, Consumer<T> handleFun) {
        RingBuffer<DisruptorMessage<T>> disruptorMq = createDisruptorMq(threadName, handleFun);
        this.disruptorMqProducer = disruptorMqProducer(disruptorMq);
    }

    private RingBuffer<DisruptorMessage<T>> createDisruptorMq(String threadName, Consumer<T> handleFun) {
        //指定事件工厂
        DisruptorEventFactory<T> factory = new DisruptorEventFactory<>();
        //指定ringbuffer字节大小，必须为2的N次方（能将求模运算转为位运算提高效率），否则将影响效率
        int bufferSize = 1024 * 256;
        //单线程模式，获取额外的性能
        Disruptor<DisruptorMessage<T>> disruptor = new Disruptor<>(factory, bufferSize, new NameThreadFactory(threadName),
                ProducerType.SINGLE, new BlockingWaitStrategy());
        DisruptorConsumer<T> disruptorConsumer = new DisruptorConsumer<>(handleFun);
        //设置事件业务处理器---消费者
        disruptor.handleEventsWith(disruptorConsumer);
        // 启动disruptor线程
        disruptor.start();
        //获取ringbuffer环，用于接取生产者生产的事件
        return disruptor.getRingBuffer();
    }

    private DisruptorMqProducer<T> disruptorMqProducer(RingBuffer<DisruptorMessage<T>> disruptorMq){
        return new DisruptorMqProducer<>(disruptorMq);
    }

    public void sendMessage(T t){
        disruptorMqProducer.send(t);
    }
}
